package org.zjt.spark.dstream

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * DStream的操作单元是Stream ，而并非RDD
  *
  * Stream 一般都是增量处理，只对增加的数据处理，不保存历史的状态。(无状态)
  *
  * updateStateByKey 显示对stream的按key做reduce操作，累加计算。（有状态reduce结果的增量累加）
  * PS: 参数1： 每次新增values   参数2：上次计算的结果
  */
object StreamWordCount extends App {
  val conf = new SparkConf().setMaster("local[2]").setAppName("StreamWordCount").set("spark.driver.allowMultipleContexts", "true")
  val ssc = new StreamingContext(conf, Seconds(10))
  val rdd = ssc.socketTextStream("192.168.83.131", 9002, StorageLevel.MEMORY_ONLY_SER)
  ssc.checkpoint("D:\\Idea workspace\\scala-demo\\src\\main\\resource\\ckeckpoint")
  val result = rdd.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)


  /*
    updateStateByKey 解释:
      以DStream中的数据进行按key做reduce操作，然后对各个批次的数据进行累加
      在有新的数据信息进入或更新时，可以让用户保持想要的任何状。使用这个功能需要完成两步：
      1) 定义状态：可以是任意数据类型
      2) 定义状态更新函数：用一个函数指定如何使用先前的状态，从输入流中的新值更新状态。
      参数1： 每次新增values   参数2：上次计算的结果
  */

  val updateFunc: (Seq[Int], Option[Int]) => Option[Int] = (values: Seq[Int], state: Option[Int]) => {
    println("values:" + values.toArray.mkString(",") + "\tstate" + state)
    val currentCount = values.sum
    val previousCount = state.getOrElse(0)
    Some(currentCount + previousCount)
  }

  def updateFunc(values: Seq[Int], state: Option[Int]) :Option[Int] ={
    println("values:" + values.toArray.mkString(",") + "\tstate" + state)
    val currentCount = values.foldLeft(0)(_+_)
    val previousCount = state.getOrElse(0)
    Some(currentCount + previousCount)
  }



  //updateStateByKey 显示对stream的按key做reduce操作，累加计算。PS: 参数1： 每次新增values   参数2：上次计算的结果
  val result2 = rdd.flatMap(_.split(" ")).map((_, 1)).updateStateByKey(updateFunc)
    .transform(_.sortByKey(false))    //对Stream中的Rdd进行操作

  result2.print()
  //result.print()
  ssc.start()
  ssc.awaitTermination()
  ssc.stop()
}
